package com.xf.day06

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import shapeless.Lazy.apply
/**
 * 广播变量
 */
object TestBroadCast {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName("debug").setMaster("local[*]")
    val sc = new SparkContext(conf)

    // 采用集合方式，实现 join
    val rdd1: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("b", 2), ("c", 3)), 2)
    val list: List[(String, Int)] = List(("a", 4), ("b", 5), ("c", 6))

    // 声明广播变量
    val broadcastList:Broadcast[List[(String, Int)]]  = sc.broadcast(list)

    val resultRDD: RDD[(String, (Int, Int))] = rdd1.map {
      case (k1, v1) => {
        var v2: Int = 0
       // for ((k3, v3) <- list.value) {
          for ((k3, v3) <- broadcastList.value) {
          if (k1 == k3) {
            v2 = v3
          }
        }
        (k1, (v1, v2))
      }
    }

    resultRDD.foreach(println)

    // 关闭链接
    sc.stop()

  }

  private def test1(): Unit = {
    val conf = new SparkConf().setAppName("debug").setMaster("local[*]")
    val sc = new SparkContext(conf)

    // 采用集合方式，实现 join
    val rdd1: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("b", 2), ("c", 3)), 2)
    val list: List[(String, Int)] = List(("a", 4), ("b", 5), ("c", 6))

    val resultRDD: RDD[(String, (Int, Int))] = rdd1.map {
      case (k1, v1) => {
        var v2: Int = 0
        for ((k3, v3) <- list.value) {
          if (k1 == k3) {
            v2 = v3
          }
        }
        (k1, (v1, v2))
      }
    }

    resultRDD.foreach(println)

    // 关闭链接
    sc.stop()
  }
}
